package com.sugon.ohdfs.integration.flink.config;

import com.sugon.ohdfs.integration.flink.job.stream.sink.*;
import com.sugon.ohdfs.integration.flink.job.stream.source.ParquetBulkStreamSourceJob;
import com.sugon.ohdfs.integration.flink.job.stream.source.RecordStreamSourceJob;
import com.sugon.ohdfs.integration.flink.service.ConsoleSink;
import com.sugon.ohdfs.integration.flink.service.TestItemSource;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;

@Configuration
public class CommonConfig {

    @Autowired
    Environment environment;

    @Bean
    public TestItemSource testItemSource(){
        return new TestItemSource();
    }

    @Bean
    public ConsoleSink consoleSink(){
        return new ConsoleSink();
    }

    @Bean
    @ConditionalOnExpression("'${jobName}'.equals('RowEncodedStreamSinkJob')")
    public RowEncodedStreamSinkJob rowEncodedStreamSinkJob(){
        return new RowEncodedStreamSinkJob();
    }

    @Bean
    @ConditionalOnExpression("'${jobName}'.equals('BulkOrcStreamSinkJob')")
    public BulkOrcStreamSinkJob bulkOrcStreamSinkJob(){
        return new BulkOrcStreamSinkJob();
    }

    @Bean
    @ConditionalOnExpression("'${jobName}'.equals('BulkParquetStreamSinkJob')")
    public BulkParquetStreamSinkJob bulkParquetStreamSinkJob(){
        return new BulkParquetStreamSinkJob();
    }

    @Bean
    @ConditionalOnExpression("'${jobName}'.equals('BulkSequenceStreamSinkJob')")
    public BulkSequenceStreamSinkJob bulkSequenceStreamSinkJob(){
        return new BulkSequenceStreamSinkJob();
    }

    @Bean
    @ConditionalOnExpression("'${jobName}'.equals('BulkAvroStreamSinkJob')")
    public BulkAvroStreamSinkJob bulkAvroStreamSinkJob(){
        return new BulkAvroStreamSinkJob();
    }

    @Bean
    @ConditionalOnExpression("'${jobName}'.equals('RowEncodedStreamSourceJob')")
    public RecordStreamSourceJob rowEncodedStreamSourceJob(){
        return new RecordStreamSourceJob();
    }

    @Bean
    @ConditionalOnExpression("'${jobName}'.equals('ParquetBulkStreamSourceJob')")
    public ParquetBulkStreamSourceJob parquetBulkStreamSourceJob(){
        return new ParquetBulkStreamSourceJob();
    }
}
